﻿using DotNetCore.CAP.Messages;
using DotNetCore.CAP.Transport;
using Microsoft.Extensions.Options;
using MQTTnet;
#if !NET8_0_OR_GREATER
using MQTTnet.Client;
using MQTTnet.Client.Disconnecting;
using MQTTnet.Client.Options;
using MQTTnet.Client.Subscribing;
#endif
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using Newtonsoft.Json;
using System.Threading.Tasks;
using System.Runtime.InteropServices;

namespace DotNetCore.CAP.MQTT
{
    public class MqttConsumerClient : IConsumerClient
    {
        private readonly SemaphoreSlim _connectionLock = new SemaphoreSlim(initialCount: 1, maxCount: 1);
        private readonly IConnectionChannelPool _connectionChannelPool;
        private readonly CAPMqttOptions _mqttOptions;
        private IMqttClient _mqttClient;
        private string _ClientId;
        public MqttConsumerClient(string ClientId, IConnectionChannelPool connectionChannelPool, IOptions<CAPMqttOptions> mqttOptions)
        {
            _connectionChannelPool = connectionChannelPool;
            _mqttOptions = mqttOptions.Value;
            _ClientId = ClientId;
        }
        public BrokerAddress BrokerAddress => new BrokerAddress("Mqtt", $"{_mqttOptions.Server}:{_mqttOptions.Port}");

        public Func<TransportMessage, object?, Task>? OnMessageCallback { get; set; }

        public Action<LogMessageEventArgs>? OnLogCallback { get; set; }
#if NETSTANDARD2_0 || NET48_OR_GREATER
        public event EventHandler<TransportMessage> OnMessageReceived;
        public event EventHandler<LogMessageEventArgs> OnLog;
#endif


        public void Subscribe(IEnumerable<string> topics)
        {
            if (topics == null)
            {
                throw new ArgumentNullException(nameof(topics));
            }

            Connect();

            foreach (var topic in topics)
            {
                _mqttClient.SubscribeAsync(new MqttClientSubscribeOptionsBuilder().WithTopicFilter(topic).Build(), CancellationToken.None).GetAwaiter().GetResult();
            }
        }
        public void Commit(object? sender)
        {
            if (_mqttClient.IsConnected)
            {
                MqttApplicationMessageReceivedEventArgs args = (MqttApplicationMessageReceivedEventArgs)sender;
                args.AcknowledgeAsync(CancellationToken.None).GetAwaiter().GetResult();
            }
        }

        public void Reject(object? sender)
        {
            throw new NotImplementedException();

        }
        public void Listening(TimeSpan timeout, CancellationToken cancellationToken)
        {
            Connect();
#if !NET8_0_OR_GREATER
            _mqttClient.UseApplicationMessageReceivedHandler(MessageReceived);
            _mqttClient.UseDisconnectedHandler(Disconnected);
#else
            _mqttClient.ApplicationMessageReceivedAsync += MessageReceived;
            _mqttClient.DisconnectedAsync += Disconnected;
#endif

            while (true)
            {
                cancellationToken.ThrowIfCancellationRequested();
                cancellationToken.WaitHandle.WaitOne(timeout);
            }
        }


        public void Connect()
        {
            if (_mqttClient != null && _mqttClient.IsConnected)
            {
                return;
            }

            _connectionLock.Wait();

            try
            {
                if (_mqttClient == null)
                {

                    _mqttClient = new
#if NET8_0_OR_GREATER
                    MqttClientFactory()
#else
                    MqttFactory()
#endif
                    .CreateMqttClient();
                    MqttClientOptionsBuilder optionsBuilder = new MqttClientOptionsBuilder();
                    // 设置服务器端地址
                    optionsBuilder.WithTcpServer(_mqttOptions.Server, _mqttOptions.Port);

                    // 设置鉴权参数
                    if (!string.IsNullOrEmpty(_mqttOptions.UserName) && !string.IsNullOrEmpty(_mqttOptions.Password))
                    {
                        optionsBuilder.WithCredentials(_mqttOptions.UserName, _mqttOptions.Password);
                    }

                    // 设置客户端序列号
                    optionsBuilder.WithClientId(_ClientId);

                    optionsBuilder.WithKeepAlivePeriod(TimeSpan.FromSeconds(2000));
                    optionsBuilder.WithCleanSession(true);
#if NET8_0_OR_GREATER
                    optionsBuilder.WithTimeout(TimeSpan.FromHours(1));
#else
                    optionsBuilder.WithCommunicationTimeout(TimeSpan.FromHours(1));
#endif
                    optionsBuilder.WithProtocolVersion(_mqttOptions.Version);
                    // 创建选项
                    var mqttClientOptions = optionsBuilder.Build();

                    _mqttClient.ConnectAsync(mqttClientOptions).GetAwaiter().GetResult();
                }
            }
            catch (Exception)
            {

                throw;
            }
            finally
            {
                _connectionLock.Release();
            }
        }

        public void Dispose()
        {
            if (_mqttClient != null)
            {
#if NET8_0_OR_GREATER
                _mqttClient.DisconnectedAsync -= Disconnected;
                _mqttClient.ApplicationMessageReceivedAsync -= MessageReceived;
#endif
                _mqttClient.Dispose();
            }
        }
        TransportMessage GetTransportMessage(string json)
        {
            IDictionary<string, string?> header;
            byte[] body;

            if (_mqttOptions.UseString)
            {
                var mqttmesage = JsonConvert.DeserializeObject<MqttMessage>(json);
                header = mqttmesage.Headers;
                body = ObjectToBytes(mqttmesage.Body);
            }
            else
            {
                var mqttmesage = JsonConvert.DeserializeObject<MqttMessageBytes>(json);
                header = mqttmesage.Headers;
                body = mqttmesage.Body;
            }

            header.Add(Headers.Group, _ClientId);

            return new TransportMessage(header, body);
        }
        public static byte[] ObjectToBytes(object obj)
        {
            string jsonString = JsonConvert.SerializeObject(obj);
            return System.Text.Encoding.UTF8.GetBytes(jsonString);
        }
#if !NET8_0_OR_GREATER
        private void MessageReceived(MqttApplicationMessageReceivedEventArgs args)
        {
            var payload = args.ApplicationMessage.Payload;
            var message = GetTransportMessage(Encoding.UTF8.GetString(payload));
            if (OnMessageCallback == null)
            {
                OnMessageReceived.Invoke(args, message);
            }
            else
                OnMessageCallback!(message, args).GetAwaiter().GetResult();
        }
        private void Disconnected(MqttClientDisconnectedEventArgs args)
        {
        }
#else
        private async Task MessageReceived(MqttApplicationMessageReceivedEventArgs args)
        {
            var payload = args.ApplicationMessage.Payload;
            var message = GetTransportMessage(Encoding.UTF8.GetString(payload));
            await OnMessageCallback!(message, args);
        }
        private async Task Disconnected(MqttClientDisconnectedEventArgs args)
        {
            await Task.CompletedTask;
        }
#endif
    }
}
